package com.niit.service

import java.text.SimpleDateFormat

import com.niit.bean.AdClickData
import com.niit.dao.AreaCountDao
import org.apache.spark.streaming.dstream.DStream


/*
需求 实时统计 每天 各地区 各城市 各广告 的点击总流量，并将其存入 MySQL。

  Key:时间、区域、城市、广告  value：总点击量
 */
class AreaCountService {

    //数据分析的方法
  def dataAnalysis(data:DStream[AdClickData]):Any={
      /*
      1.  将Kakfa中的数据提取出 时间戳、区域、城市、广告
       */
    val resduceDS = data.map(line=>{
      val ts: String = line.ts
      val format = new SimpleDateFormat("yyyy-MM-dd")
      val day = format.format(new java.util.Date(ts.toLong))
      val area = line.area
      val city = line.city
      val ad = line.ad
      ( (day,area,city,ad) ,1 )
    })

    //2. 对每天每个区域每个城市点击每广告 计算 总的点击量
    val resDS: DStream[((String, String, String, String), Int)] = resduceDS.reduceByKey(_ + _)

    //3.存入数据库
    resDS.foreachRDD(rdd=>{
      rdd.foreach{
        case ( (day,area,city,ad),count)=>{
          //TODO   存入数据库
          println(s"${day},${area},${city},${ad},${count}")
          val areaCountDao = new AreaCountDao
          areaCountDao.insertAreaCount(day,area,city,ad,count);
        }
      }
    })

  }

}
